package com.lzy.canal.register.consumer;

import com.lzy.canal.component.consumer.mysql.DefaultDdlConsumer;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * ddl 消息处理器
 */
@Slf4j
public class DdlConsumerRegister implements ConsumerRegister<DefaultDdlConsumer> {

    private final DefaultDdlConsumer defaultDdlConsumer;

    private Map<String, DefaultDdlConsumer> registerConsumerMap = new ConcurrentHashMap<>();

    public DdlConsumerRegister(DefaultDdlConsumer defaultDdlConsumer) {
        this.defaultDdlConsumer = defaultDdlConsumer;
    }

    @Override
    public ConsumerRegister add(String schemaName, String tableName, DefaultDdlConsumer mysqlConsumer) {
        registerConsumerMap.put(getKey(schemaName, tableName), mysqlConsumer);
        return this;
    }

    @Override
    public DefaultDdlConsumer getConsumer(String schemaName, String tableName) {
        DefaultDdlConsumer consumer = registerConsumerMap.get(getKey(schemaName, tableName));
        if (consumer == null) {
            log.info("未找到消息处理组件,使用默认消息处理组件");
            return defaultDdlConsumer;
        }
        return consumer;
    }
}
